package com.bbx.flink.table_api_or_sql.table_api_all;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Joins
 * https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/tableApi.html#over-windows
 */
public class Demo2 {


    private static StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    private static StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

    public static void main(String[] args) throws Exception {

        env.setParallelism(1);
        String createLeftTable = "CREATE TABLE leftTable (\n" +
                " id String,name String,tel String " +
                ") WITH (\n" +
                "  'connector.type' = 'filesystem',\n" +
                "  'connector.path' = 'file:///C:\\Users\\kaifacs\\Desktop\\33.csv', \n" +
                "  'format.type' = 'csv' \n" +
                ")";

        String createRightTable = "CREATE TABLE rightTable (\n" +
                " code String,age String,address String " +
                ") WITH (\n" +
                "  'connector.type' = 'filesystem',\n" +
                "  'connector.path' = 'file:///C:\\Users\\kaifacs\\Desktop\\44.csv', \n" +
                "  'format.type' = 'csv' \n" +
                ")";

        tableEnv.executeSql(createLeftTable);
        tableEnv.executeSql(createRightTable);

        Table leftTable = tableEnv.from("leftTable");
        Table rightTable = tableEnv.from("rightTable");

        int flag = 1;
        switch (flag) {
            case 0:
                InnerJoin(leftTable, rightTable);
                break;
            case 1:
                outerJoin(leftTable, rightTable);
                break;
            case 2:
                InnerOrOuterIntervalJoin();
                break;
        }

        env.execute();
    }

    /**
     * 类似与SQL 中的join，所有表的字段必须唯一（表字段不能相同）
     * Similar to a SQL JOIN clause.
     * Joins two tables. Both tables must have distinct field names and at least one equality join predicate
     * must be defined through join operator or using a where or filter operator.
     *
     * @param leftTable
     * @param rightTable
     */
    static void InnerJoin(Table leftTable, Table rightTable) {
        Table innerJoinTable = leftTable.join(rightTable)
                .where($("id").isEqual($("code")))
                .select($("id"), $("code"), $("name"), $("tel"), $("age"), $("address"));
        tableEnv.toAppendStream(innerJoinTable, Row.class).print("innerJoinTable");
    }


    /**
     * 个人理解--根据结果猜想（不一定正确）
     * 1、当两边表都存在有对方不存在的数据时，
     * 即：left 表有的数据 right 表可能部分不存在，或者相反，此时会以主表和 null 表合并，然后在更新数据
     * 2、关联表中完全包含主表数据，此时会直接合并
     * Similar to SQL LEFT/RIGHT/FULL OUTER JOIN clauses. Joins two tables. Both tables must have distinct field names
     * and at least one equality join predicate must be defined.
     */
    static void outerJoin(Table leftTable, Table rightTable) {
        Table leftOuterJoinTable = leftTable.leftOuterJoin(rightTable, $("id").isEqual($("code")) )
                .select($("id"),$("name"),$("tel"),$("code"),$("age"),$("address"));
        tableEnv.toRetractStream(leftOuterJoinTable, Row.class).print("leftOuterJoinTable");

        Table rightOuterJoinTable = leftTable.rightOuterJoin(rightTable, $("id").isEqual($("code")))
                .select($("*"));
        tableEnv.toRetractStream(rightOuterJoinTable,Row.class).print("rightOuterJoinTable");

        Table fullOuterJoinTable = leftTable.fullOuterJoin(rightTable, $("id").isEqual($("code")))
                .select($("id"), $("name"), $("tel"), $("code"), $("age"), $("address"));
        tableEnv.toRetractStream(fullOuterJoinTable,Row.class).print("fullOuterJoinTable");



    }


    static  void InnerOrOuterIntervalJoin(){
        String createLeft = "CREATE TABLE left (\n" +
                " id String,name String,tel String ,leftProctime   as proctime() " +
                ") WITH (\n" +
                "  'connector.type' = 'filesystem',\n" +
                "  'connector.path' = 'file:///C:\\Users\\kaifacs\\Desktop\\33.csv', \n" +
                "  'format.type' = 'csv' \n" +
                ")";

        String createRight = "CREATE TABLE right (\n" +
                " code String,age String,address String  ,rightProcetime as proctime()" +
                ") WITH (\n" +
                "  'connector.type' = 'filesystem',\n" +
                "  'connector.path' = 'file:///C:\\Users\\kaifacs\\Desktop\\44.csv', \n" +
                "  'format.type' = 'csv' \n" +
                ")";

        tableEnv.executeSql(createLeft);
        tableEnv.executeSql(createRight);

        Table left = tableEnv.from("left");
        Table right = tableEnv.from("right");

//        left.join(right)
//                .where(and($("")))








    }
}
